package Log

import (
	"back_go/DAO"
	models "back_go/DAO/model"

	"context"
	"encoding/json"
	"fmt"
	"github.com/apache/rocketmq-client-go/v2"
	"github.com/apache/rocketmq-client-go/v2/consumer"
	"github.com/apache/rocketmq-client-go/v2/primitive"
)

func ConsumeMessages() {
	c, err := rocketmq.NewPushConsumer(
		consumer.WithGroupName("your_consumer_group"),
		consumer.WithNsResolver(primitive.NewPassthroughResolver([]string{"192.168.3.242:9876"})),
	)
	fmt.Println("定时任务")
	if err != nil {
		fmt.Println("创建消费者失败：", err)
		return
	}
	var engine = DAO.Engine

	err = c.Subscribe("AdminLogs", consumer.MessageSelector{}, func(ctx context.Context, msgs ...*primitive.MessageExt) (consumer.ConsumeResult, error) {
		for _, msg := range msgs {
			var temp map[string]interface{}
			_ = json.Unmarshal(msg.Body, &temp)
			tsFloat, _ := temp["ts"].(float64)
			var logObj models.Logs
			logObj.Timestamp = int64(tsFloat)
			logObj.LogLevel = temp["level"].(string)
			logObj.Caller = temp["caller"].(string)
			logObj.Msg = temp["msg"].(string)
			logObj.Method = temp["method"].(string)
			logObj.Path = temp["path"].(string)
			logObj.Query = temp["query"].(string)
			logObj.Body = temp["body"].(string)
			logObj.ApplyIp = temp["ip"].(string)
			_, err := engine.Insert(&logObj)
			if err != nil {
				return 0, err
			}
			fmt.Println(msgs)
		}
		return consumer.ConsumeSuccess, nil
	})
	if err != nil {
		fmt.Println("订阅主题失败：", err)
		return
	}

	err = c.Start()
	if err != nil {
		fmt.Println("启动消费者失败：", err)
		return
	}

	defer c.Shutdown()
}
